package org.zfes.snowier.zbss.xzy.store.queue;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.zfes.snowier.zbss.xzy.store.repository.api.DataRepository;
import org.zfes.snowier.zbss.xzy.store.repository.api.StoreDataMsg;

import com.esotericsoftware.minlog.Log;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
public class PersistManager  implements InitializingBean , DisposableBean{

	private Logger logger = LoggerFactory.getLogger(this.getClass());
	
	@Autowired DataRepository dataRepository;
	
//	private static BufferQueue queue = new BufferQueue();
	
	private static  PersisitBlockQueue queue=new PersisitBlockQueue();
	
//    private final ReadWriteLock lock = new ReentrantReadWriteLock();
//    private final Lock readLock = lock.readLock();
//    private final Lock writeLock = lock.writeLock();
    
    private int bufferSize=500;
    
    private int persistInternal=2;
    
    
	 static final int maxPersistThreads=10;
	 static final int maxPersistThreadQueueSize=4096*3;
	 
	 
	private static ExecutorService offerExecutorService;
	
	private static ScheduledExecutorService scheduledPostExecutorService;
	
	private static ExecutorService persisitExecutorService;
	
	 
	private volatile static AtomicBoolean sheduelPoolStarted = new AtomicBoolean(false);
	
	private volatile static AtomicBoolean stopAcceptData = new AtomicBoolean(false);

	
	@Override
	public void destroy() throws Exception {
		stopAcceptData.set(true);
		//flush();
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		startTimerSheduel();
	}
	
	
	public  void offer(StoreDataMsg msg) {
		if(stopAcceptData.get()) {
			return;
		}
//		getOfferPool().submit(()->{
			 //this.writeLock.lock();
			try {
				queue.offer(msg);
				logger.info("-after-offer-queue-size:{}",queue.size());
			} finally {
		     // this.writeLock.unlock();
		   }
			
			if(queue.size()>=bufferSize) {
				this.persisitBatch();
			}
			
//		});
	}
	
	public void flush() {
		logger.info("-before- flush-queue-size:{}",queue.size());
		while(!queue.isEmpty()){
			persisitBatch();
		}
		logger.info("-after- flush-queue-size:{}",queue.size());
	}
	
	
	private void persisitBatch(){
		
		getPersisitPool().submit(()->{
			// this.readLock.lock();
				try {
					//logger.info("-before--persisitBatch- -size:{}",queue.size());
					
					List<Object> dataList = queue.poll(bufferSize);
					if(dataList.isEmpty()) {
						return;
					}
					try {
						dataRepository.batchSave("mo_data_test", dataList.stream().map(v->(StoreDataMsg)v).collect(Collectors.toList()));
						logger.info("-after--persisitBatch- -queue-size:{},-dataList-size:{}",queue.size(),dataList.size());
					}catch (Exception e) {
						Log.error("-batchSave-error:", e);
					}
					
				} finally {
			    //  this.readLock.unlock();
			  }
		});
		
	}
	
	private  void startTimerSheduel(){
		if(sheduelPoolStarted.get()) {
			return;
		}
		
		if(scheduledPostExecutorService==null) {
			scheduledPostExecutorService=Executors.newSingleThreadScheduledExecutor();
		}
		
		scheduledPostExecutorService.scheduleAtFixedRate(()->{
			try {
				
				if(queue.size() < bufferSize) {
					
					this.persisitBatch();
					//logger.info("-after-TimerSheduel -size:{}",queue.size());
				}
				
			}catch (Exception e) {
				e.printStackTrace();
			}
			
		},1,persistInternal,TimeUnit.SECONDS);
		
		sheduelPoolStarted.set(true);
	}
	
	
	
	private static ExecutorService  getOfferPool(){
//		if(offerExecutorService==null) {
//			offerExecutorService=Executors.newFixedThreadPool(5);
//		}
//		return offerExecutorService;
		
		if(offerExecutorService != null) {
			return offerExecutorService;
		}
		ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("-offer-Pool-pool-%d").build();
		persisitExecutorService = new ThreadPoolExecutor(5, 10,0L, 
						TimeUnit.MILLISECONDS,
						new LinkedBlockingQueue<Runnable>(100), 
						namedThreadFactory, 
						new ThreadPoolExecutor.AbortPolicy());
		return persisitExecutorService;	
		
	}
	private static ExecutorService  getPersisitPool() {
		if(persisitExecutorService != null) {
			return persisitExecutorService;
		}
		ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("-persist-Pool-pool-%d").build();
		persisitExecutorService = new ThreadPoolExecutor(5, maxPersistThreads,0L, 
						TimeUnit.MILLISECONDS,
						new LinkedBlockingQueue<Runnable>(maxPersistThreadQueueSize), 
						namedThreadFactory, 
						new ThreadPoolExecutor.AbortPolicy());
		return persisitExecutorService;				
		
	}


	
}
